package com.doit.spark.day04.demo

import com.doit.spark.day01.utils.SparkUtil
import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.{JdbcRDD, RDD}

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}
import scala.collection.mutable

/**
 * @DATE 2022/1/6/14:40
 * @Author MDK
 * @Version 2021.2.2
 * */
object D01_Demo01 {
  def main(args: Array[String]): Unit = {
    val sc: SparkContext = SparkUtil.getSc
    /**
     * 每个人的总金额  不用关联mysql中的数据
     */
    val fileDataRDD: RDD[String] = sc.textFile("data/demo/demo01.txt")

    val uidAndMoneyRDD: RDD[(String, Double)] = fileDataRDD.map(line => {
      val arr: Array[String] = line.split(",")
      (arr(1), arr(2).toDouble)
    })
    // groupBy  groupByKey可以 , 效率没有reduceByKey高 , 支持Map端聚合
    val uidSumMoney: RDD[(String, Double)] = uidAndMoneyRDD.reduceByKey(_ + _)
    //uidSumMoney.foreach(println)
    // uidSumMoney.saveAsTextFile("data/demo1_out1")
    /**
     * 每个订单的信息关联mysql用户名
     * 方式1   获取每个订单的uid 去mysql中匹配   map算子  不好  连接次数过多
     * 方式2  mapPartitions    一个分区获取一个连接    处理分区内的每条数据  进行拼接
     *
     */
    /* fileDataRDD.map(line=>{
       val uid: String = line.split(",")(1)
       // 连接mysql  获取用户信息
     })*/
    val resRDD1: RDD[String] = fileDataRDD.mapPartitions(iters => {
      // 获取连接
      val conn: Connection = DriverManager.getConnection("jdbc:mysql://linux01:3306/doit", "root", "root")
      val ps: PreparedStatement = conn.prepareStatement("select * from tb_user where uid = ?")
      // 遍历数据  获取用户
      iters.map(line => {
        //获取信息
        val uid: String = line.split(",")(1) //订单中的uid
        ps.setString(1, uid) // 预编译SQL
        val rs: ResultSet = ps.executeQuery() //执行sql语句  返回结果
        rs.next() // 获取数据
        val name: String = rs.getString("name") // 获取行数据中的name属性
        line + "," + name
      })
    })
    // resRDD1.saveAsTextFile("data/demo1_out2")
    /**
     * 方式三 :  将数据库中的数据转换成RDD   join
     * 1) 本地集合并行化  2) 加载外部文件  3) 转换算子  4) JDBCRDD
     */
    /*
    class JdbcRDD[T: ClassTag](
    sc: SparkContext,
    getConnection: () => Connection,
    sql: String,  ... where id >= ?  and id <= ?
    lowerBound: Long,  第一个占位符   1
    upperBound: Long,  第二个占位符   3
    numPartitions: Int,
    mapRow: (ResultSet) => T = JdbcRDD.resultSetToObjectArray _)
    extends RDD[T](sc, Nil) with Logging {
     */
    //--------------------------------生成JDBCRDD----------------------------------------
    // 获取连接
    /*val conn = () => DriverManager.getConnection("jdbc:mysql://linux01:3306/doit", "root", "root")
    // 执行SQL语句  表要有一个主键  自增
    val sql = "select * from tb_user where id >= ?  and id <= ?"
    // 封装结果
    val row = (rs: ResultSet) => {
      val uid: String = rs.getString("uid")
      val name: String = rs.getString("name")
      (uid, name)
    }
    val mysqlTableRDD = new JdbcRDD[(String, String)](sc, conn, sql, 1, 3, 2, row)
    //--------------------------------------------------------------------------------------------------

    val fileDataRDD2: RDD[String] = sc.textFile("data/demo/")
    val fileRDD: RDD[(String, String)] = fileDataRDD2.map(line => {
      val arr: Array[String] = line.split(",")
      (arr(1), line)
    })
    fileRDD.foreach(println)
    //----------------------------------------------------------------------------
    val res: RDD[(String, (String, String))] = mysqlTableRDD.join(fileRDD)
    res.map(tp => {
      tp._2._1 + "," + tp._2._2
    }).foreach(println)*/


    /**
     * 方式四 :  Driver端连接数据库 获取所有的用户信息   广播  闭包
     */
    val connection = DriverManager.getConnection("jdbc:mysql://linux01:3306/doit", "root", "root")
    val ps: PreparedStatement = connection.prepareStatement("select * from tb_user")
    val rs: ResultSet = ps.executeQuery()

      //封装用户数据
      val userInfo = new mutable.HashMap[String, String]()
      while (rs.next()){
        val uid = rs.getString("uid")
        val name = rs.getString("name")
        userInfo.put(uid, name)
      }
    //将从mysql中获取的用户信息广播出去
    val buser: Broadcast[mutable.HashMap[String, String]] = sc.broadcast(userInfo)
    val resRDD2: RDD[String] = fileDataRDD.map(line => {
      val uid = line.split(",")(1)
      //获取广播变量的数据
      val mp: mutable.HashMap[String, String] = buser.value
      val name = mp.getOrElse(uid, "未知用户")
      line + "," + name
    })
    resRDD2.foreach(println)
    /**
     * 每个人的总金额  不用关联mysql中的用户名
     */
      println("-----------------------------------------------------------------------------")
    val resRDD3 = resRDD2.map(line => {
      val arr = line.split(",")
      (arr(3), arr(2).toDouble)
    }).reduceByKey(_ + _)
    resRDD3.foreach(println)


  }
}
